using Confluent.Kafka;
using JT808Protocol.Abstracts;
using JT808Protocol.Models;
using System;
using System.Collections.Generic;
using System.Net;
using System.Threading;
using System.Threading.Tasks;

namespace JT808Gateway.WebHost
{
    /// <summary>
    /// Kafka转发器
    /// </summary>
    public class KafkaForwarder : IDataForwarder
    {
        private KafkaConfig _kafkaConfig;

        private ProducerConfig _producerConfig;

        private Queue<JT808Package> _queue;

        private Task _taskForward;

        public KafkaForwarder(KafkaConfig config)
        {
            _kafkaConfig = config;

            _producerConfig = new ProducerConfig
            {
                BootstrapServers = $"{_kafkaConfig.Host}:{_kafkaConfig.Port}",
                ClientId = Dns.GetHostName()
            };

            _queue = new Queue<JT808Package>();

            _taskForward = Task.Factory.StartNew(async () =>
            {
                using var producer = new ProducerBuilder<string, JT808Package>(_producerConfig).Build();
                while (true)
                {
                    if (_queue.Count > 0)
                    {
                        try
                        {
                            var package = _queue.Dequeue();

                            //Todo : 转发消息供微服务消费
                            var dr = await producer.ProduceAsync(_kafkaConfig.JT808GwSendTopic, new Message<string, JT808Package>
                            {
                                Key = $"{package.Header.MessageID}",
                                Value = package
                            });
                            Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
                        }
                        catch (ProduceException<Null, string> e)
                        {
                            Console.WriteLine($"Delivery failed: {e.Error.Reason}");
                        }
                    }
                    else
                        Thread.Sleep(100);
                }
            });
        }

        /// <summary>
        /// 使用Kafka转发消息
        /// </summary>
        /// <param name="package"></param>
        /// <returns></returns>
        public async Task SendAsync(JT808Package package)
        {
            if (_queue.Count >= 5000)
                _queue.Clear();

            _queue?.Enqueue(package);
        }
    }
}